package com.hh.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.hh.constant.QueueConstant;
import com.hh.entity.Prod;
import com.hh.es.ProdEs;
import com.hh.dao.ProdEsMapper;
import com.hh.service.ImportService;
import com.hh.service.ProdService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

/**
 * @author 辉辉 2022/8/18 8:45
 */
@Service
@Slf4j
public class ImportServiceImpl implements ImportService, CommandLineRunner {
    @Resource
    private ProdEsMapper prodEsMapper;

    @Resource
    private ProdService prodService;
    /**
     * 给一个初始时间
     */
    private Date t1;

    /**
     * 每次导入的条数
     */
    @Value("${esimport.size}")
    private Integer size;

    /**
     * 全量导入
     * mysql 的prod导入es
     */
    @Override
    public void importAll() {
        log.info("开始全量导入");
        // 计算分页
        // 查询到所有的count
        Integer totalCount = prodService.getTotalCount(null, null);
        if (totalCount == null || totalCount == 0) {
            // 根本就没有数据导入
            log.warn("根本就没有数据导入");
            return;
        }
        // 计算分页 totalCount / size
        long page = totalCount % size == 0 ? totalCount / size : ((totalCount / size) + 1);
        // 循环
        for (int i = 1; i <= page; i++) {
            // 分页 查询 导入 时间
            importToEs(i, size, null, null);
        }
        log.info("全量导入完成");
        // 查询到一个List集合
//         prodEsMapper.saveAll()
        // 记录一个时间 初始时间
        t1 = new Date();
    }

    /**
     * 导入es的方法
     * @param current
     * @param size
     * @param t1
     * @param t2
     */
    private void importToEs(int current, Integer size, Date t1, Date t2) {
        // 1. 分页查询
        // 组装一个page 如果让他做分页 查询总条数 这是个小优化
        Page<Prod> page = new Page<>(current, size, false);
        Page<Prod> prodPage = prodService.findProdByPageToEs(page, t1, t2);
        List<Prod> prodList = prodPage.getRecords();
        // new 一个空的集合 专门放导入es的对象
        List<ProdEs> prodEsList = new ArrayList<>(prodList.size() * 2);
        // 2. 组装数据
        prodList.forEach(prod -> {
            ProdEs prodEs = new ProdEs();
            // 对象拷贝 只要是属性名称相同 就可以拷贝 第三个参数是忽略大小写
            BeanUtil.copyProperties(prod, prodEs, true);
            // 放入new的空集合
            prodEsList.add(prodEs);
        });
        // 3. 导入
        // 批量导入
        prodEsMapper.saveAll(prodEsList);
    }

    /**
     * 增量导入 定时任务
     * 商家上架一个商品 定时任务 30 min 去查询数据库mysql 如果有修改的就查出来 导入到es 跟es做同步
     */
    @Override
    // 第一个参数是项目启动完首次执行定时任务的时间
    // 第二个参数是执行完第一次定时任务后每次间隔多长时间
    @Scheduled(initialDelay = 3 * 60 * 1000, fixedRate = 3 * 60 * 1000)
    public void importUpdate() {
        log.info("增量导入开始");
        // 这个t2比t1大30分钟
        Date t2 = new Date();
        log.error("t1是：{}, t2是：{}", t1, t2);
        // 先查询数量 也要做分页
        Integer totalCount = prodService.getTotalCount(t1, t2);
        if (totalCount == null || totalCount == 0) {
            // 根本就没有数据导入
            log.warn("根本就没有数据导入");
            // 缩小时间锁定 把时间窗口锁定
            t1 = t2;
            return;
        }
        // 计算分页 totalCount / size
        long page = totalCount % size == 0 ? totalCount / size : ((totalCount / size) + 1);
        // 循环
        for (int i = 1; i <= page; i++) {
            // 分页 查询 导入 时间
            importToEs(i, size, t1, t2);
        }
        // 优化
        // 代码走到这里来了
        // 被修改过了 数据
        // 时间窗口滑动
        t1 = t2;
        log.info("增量导入结束");
    }

    /**
     * 定义一个空方法
     */
    @Override
    public void quickImport() {
    }

    /**
     * 快速导入 用户下单减库存了 mq实现
     * 用户下单 有一个order-service 我们做一个解耦合
     * 用户下订单量特别大 如果使用openfeign 对服务器压力很大 而且他性能不好
     * 使用mq 特点：异步，解耦合，削峰，限流，解耦合
     * 业务流程：
     *  1. 用户下订单，会修改库存，mysql 同时同步es
     *  2. order-service 把减库存的消息放入mq prodId count Map<Long, Integer> 1 -2
     *  3. 处理消息
     */
    @RabbitListener(queues = QueueConstant.PROD_CHANGE_QUEUE, concurrency = "3-5")
    public void quickImportByMq(Message message, Channel channel) {
        // 拿到消息体
        String msgStr = new String(message.getBody());
        // 把数据转成map集合
        JSONObject prodInfo = JSON.parseObject(msgStr);
        // 消费消息
        List<Long> prodIds = new ArrayList<>();
        prodInfo.forEach((prodId, count) -> {
            // 根据id查询es 查到要修改的对象
            prodIds.add(Long.valueOf(prodId));
        });
        // 查询es
        Iterable<ProdEs> prodChangeList = prodEsMapper.findAllById(prodIds);
        // 循环修改数据
        prodChangeList.forEach(prodEs -> {
            Long prodId = prodEs.getProdId();
            // 改变的数量
            Integer count = prodInfo.getInteger(String.valueOf(prodId));
            // 操作数量的改变 方便做增加和减少的
            long finalCount = prodEs.getTotalStocks() + count;
            if (finalCount < 0) {
                log.error("修改es的库存小于0了");
                throw new IllegalArgumentException(prodId + "的es库存小于0");
            }
            prodEs.setTotalStocks(finalCount);
        });
        // 插入es
        prodEsMapper.saveAll(prodChangeList);
        // 签收消息
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            // 只签收自己的
            channel.basicAck(deliveryTag, false);
            log.info("同步es成功");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     *  当 ioc 容器初始化完成 当boot启动成功 就会执行
     * @param args
     * @throws Exception
     */
    @Override
    public void run(String... args) throws Exception {
        importAll();
    }
}
